/**
 * Copyright 2019 Anthony Trinh
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package ch.qos.logback.core;

import ch.qos.logback.core.spi.AppenderAttachable;
import ch.qos.logback.core.spi.AppenderAttachableImpl;
import ch.qos.logback.core.util.InterruptUtil;

import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * This appender and derived classes, log events asynchronously.  In order to avoid loss of logging events, this
 * appender should be closed. It is the user's  responsibility to close appenders, typically at the end of the
 * application lifecycle.
 * <p>
 * This appender buffers events in a {@link BlockingQueue}. {@link ch.qos.logback.core.AsyncAppenderBase.Worker} thread created by this appender takes
 * events from the head of the queue, and dispatches them to the single appender attached to this appender.
 * <p>
 * Please refer to the <a href="http://logback.qos.ch/manual/appenders.html#AsyncAppender">logback manual</a> for
 * further information about this appender.</p>
 *
 * @param <E> type of object to log
 * @author Ceki G&uuml;lc&uuml;
 * @author Torsten Juergeleit
 * @since 1.0.4
 */
public class AsyncAppenderBase<E> extends UnsynchronizedAppenderBase<E> implements AppenderAttachable<E> {

  AppenderAttachableImpl<E> aai = new AppenderAttachableImpl<E>();
  BlockingQueue<E> blockingQueue;

  /**
   * The default buffer size.
   */
  public static final int DEFAULT_QUEUE_SIZE = 256;
  int queueSize = DEFAULT_QUEUE_SIZE;

  int appenderCount = 0;

  static final int UNDEFINED = -1;
  int discardingThreshold = UNDEFINED;
  boolean neverBlock = false;

  Worker worker = new Worker();

  /**
   * The default maximum queue flush time allowed during appender stop. If the
   * worker takes longer than this time it will exit, discarding any remaining
   * items in the queue
   */
  public static final int DEFAULT_MAX_FLUSH_TIME = 1000;
  int maxFlushTime = DEFAULT_MAX_FLUSH_TIME;

  /**
   * Is the eventObject passed as parameter discardable? The base class's implementation of this method always returns
   * 'false' but sub-classes may (and do) override this method.
   * <p>
   * Note that only if the buffer is nearly full are events discarded. Otherwise, when the buffer is "not full"
   * all events are logged.
   *
   * @param eventObject the log event
   * @return - true if the event can be discarded, false otherwise
   */
  protected boolean isDiscardable(E eventObject) {
    return false;
  }


  /**
   * Pre-process the event prior to queueing. The base class does no pre-processing but sub-classes can
   * override this behavior.
   *
   * @param eventObject the log event
   */
  protected void preprocess(E eventObject) {
  }


  @Override
  public void start() {
    if (isStarted()) {
      return;
    }
    if (appenderCount == 0) {
      addError("No attached appenders found.");
      return;
    }
    if (queueSize < 1) {
      addError("Invalid queue size [" + queueSize + "]");
      return;
    }
    blockingQueue = new ArrayBlockingQueue<E>(queueSize);

    if (discardingThreshold == UNDEFINED)
      discardingThreshold = queueSize / 5;
    addInfo("Setting discardingThreshold to " + discardingThreshold);
    worker.setDaemon(true);
    worker.setName("AsyncAppender-Worker-" + getName());
    // make sure this instance is marked as "started" before staring the worker Thread
    super.start();
    worker.start();
  }

  @Override
  public void stop() {
    if (!isStarted())
      return;

    // mark this appender as stopped so that Worker can also processPriorToRemoval if it is invoking aii.appendLoopOnAppenders
    // and sub-appenders consume the interruption
    super.stop();

    // interrupt the worker thread so that it can terminate. Note that the interruption can be consumed
    // by sub-appenders
    worker.interrupt();

    InterruptUtil interruptUtil = new InterruptUtil(context);

    try {
      interruptUtil.maskInterruptFlag();

      worker.join(maxFlushTime);

      //check to see if the thread ended and if not add a warning message
      if (worker.isAlive()) {
        addWarn("Max queue flush timeout (" + maxFlushTime + " ms) exceeded. " + blockingQueue.size() +
        " queued events were possibly discarded.");
      } else {
        addInfo("Queue flush finished successfully within timeout.");
      }
    } catch (InterruptedException e) {
      int remaining = blockingQueue.size();
      addError("Failed to join worker thread. " + remaining + " queued events may be discarded.", e);
    } finally {
      interruptUtil.unmaskInterruptFlag();
    }
  }


  @Override
  protected void append(E eventObject) {
    if (isQueueBelowDiscardingThreshold() && isDiscardable(eventObject)) {
      return;
    }
    preprocess(eventObject);
    put(eventObject);
  }

  private boolean isQueueBelowDiscardingThreshold() {
    return (blockingQueue.remainingCapacity() < discardingThreshold);
  }

  private void put(E eventObject) {
    if (neverBlock) {
      blockingQueue.offer(eventObject);
    } else {
      putUninterruptibly(eventObject);
    }
  }

  private void putUninterruptibly(E eventObject) {
    boolean interrupted = false;
    try {
      while (true) {
        try {
          blockingQueue.put(eventObject);
          break;
        } catch (InterruptedException e) {
          interrupted = true;
        }
      }
    } finally {
      if (interrupted) {
        Thread.currentThread().interrupt();
      }
    }
  }

  public int getQueueSize() {
    return queueSize;
  }

  public void setQueueSize(int queueSize) {
    this.queueSize = queueSize;
  }

  public int getDiscardingThreshold() {
    return discardingThreshold;
  }

  public void setDiscardingThreshold(int discardingThreshold) {
    this.discardingThreshold = discardingThreshold;
  }

  public int getMaxFlushTime() {
    return maxFlushTime;
  }

  public void setMaxFlushTime(int maxFlushTime) {
    this.maxFlushTime = maxFlushTime;
  }

  /**
   * Returns the number of elements currently in the blocking queue.
   *
   * @return number of elements currently in the queue.
   */
  public int getNumberOfElementsInQueue() {
    return blockingQueue.size();
  }

  public void setNeverBlock(boolean neverBlock) {
    this.neverBlock = neverBlock;
  }

  public boolean isNeverBlock() {
    return neverBlock;
  }

  /**
   * The remaining capacity available in the blocking queue.
   *
   * @return the remaining capacity
   * See {@link java.util.concurrent.BlockingQueue#remainingCapacity}
   */
  public int getRemainingCapacity() {
    return blockingQueue.remainingCapacity();
  }



  public void addAppender(Appender<E> newAppender) {
    if (appenderCount == 0) {
      appenderCount++;
      addInfo("Attaching appender named ["+newAppender.getName()+"] to AsyncAppender.");
      aai.addAppender(newAppender);
    } else {
      addWarn("One and only one appender may be attached to AsyncAppender.");
      addWarn("Ignoring additional appender named [" + newAppender.getName() + "]");
    }
  }

  public Iterator<Appender<E>> iteratorForAppenders() {
    return aai.iteratorForAppenders();
  }

  public Appender<E> getAppender(String name) {
    return aai.getAppender(name);
  }

  public boolean isAttached(Appender<E> eAppender) {
    return aai.isAttached(eAppender);
  }

  public void detachAndStopAllAppenders() {
    aai.detachAndStopAllAppenders();
  }

  public boolean detachAppender(Appender<E> eAppender) {
    return aai.detachAppender(eAppender);
  }

  public boolean detachAppender(String name) {
    return aai.detachAppender(name);
  }

  class Worker extends Thread {

    public void run() {
      AsyncAppenderBase<E> parent = AsyncAppenderBase.this;
      AppenderAttachableImpl<E> aai = parent.aai;

      // loop while the parent is started
      while (parent.isStarted()) {
        try {
          E e = parent.blockingQueue.take();
          aai.appendLoopOnAppenders(e);
        } catch (InterruptedException ie) {
          break;
        }
      }

      addInfo("Worker thread will flush remaining events before exiting.");
      for (E e : parent.blockingQueue) {
        aai.appendLoopOnAppenders(e);
        parent.blockingQueue.remove(e);
      }

      aai.detachAndStopAllAppenders();
    }
  }
}
